Commit 020fe052751709b1c87c522852e218c283652c4c
Parents : 0e86b57
Author : Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-05-11T18:24:22-05:00
feat(backend) add memory diagnostics, limit sqlite cache statements
Changes
5 files changed, 752 insertions(+), 4 deletions(-)
Diff
diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py
index ec1122f3..eb658ebb 100644
--- a/meshchatx/meshchat.py
+++ b/meshchatx/meshchat.py
@@ -315,8 +315,11 @@ class ReticulumMeshChat:
ssl_key_path: str | None = None,
rns_loglevel: str | None = None,
migration_context: dict | None = None,
+ memory_diag_enabled: bool = False,
):
self.running = True
+ self._memory_diag_enabled = memory_diag_enabled
+ self._mem_diag = None
self.migration_context = (
migration_context if migration_context is not None else {}
)
@@ -2305,16 +2308,45 @@ class ReticulumMeshChat:
if gc_counter >= 300:
gc_counter = 0
sweep_stale_links()
- gc.collect()
+ # Python 3.14+ incremental GC: with threshold[2]==0 full gen2
+ # collections are never scheduled automatically, so force one.
+ if sys.version_info >= (3, 14) and gc.get_threshold()[2] == 0:
+ gc.collect(2)
+ else:
+ gc.collect()
await asyncio.sleep(1)
+ async def _memory_diag_snapshot_loop(self):
+ if not self._mem_diag:
+ return
+ while self.running and self._mem_diag.enabled:
+ try:
+ await asyncio.to_thread(self._mem_diag.snapshot)
+ n = len(self._mem_diag._snapshots)
+ if n % 12 == 0:
+ report = await asyncio.to_thread(
+ self._mem_diag.diff_snapshots,
+ top_n=10,
+ )
+ if report:
+ growth = sum(r["size_mib"] for r in report)
+ print(
+ f"[mem_diag] Snapshot #{n}: +{growth:.2f} MiB "
+ f"growth in top {len(report)} sites",
+ )
+ except Exception as exc:
+ print(f"[mem_diag] Snapshot error: {exc}")
+ await asyncio.sleep(300) # every 5 minutes
+
# automatically syncs propagation nodes based on user config
async def announce_sync_propagation_nodes(self, session_id, context=None):
ctx = context or self.current_context
if not ctx:
return
+ router = ctx.message_router
+ sync_start_time = None
while self.running and ctx.running and ctx.session_id == session_id:
auto_sync_interval_seconds = ctx.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.get()
last_synced_at = (
@@ -2329,9 +2361,30 @@ class ReticulumMeshChat:
)
# sync
- if should_sync:
+ if should_sync and sync_start_time is None:
+ sync_start_time = time.monotonic()
await self.sync_propagation_nodes(context=ctx)
+ # stuck-sync watchdog and completion detection
+ if sync_start_time is not None and router:
+ state = router.propagation_transfer_state
+ pr_complete = getattr(router, "PR_COMPLETE", None)
+ if state in {router.PR_IDLE, pr_complete}:
+ ctx.config.lxmf_preferred_propagation_node_last_synced_at.set(
+ int(time.time())
+ )
+ await self.send_config_to_websocket_clients(context=ctx)
+ sync_start_time = None
+ elif time.monotonic() - sync_start_time > 120:
+ self.stop_propagation_node_sync(context=ctx)
+ with contextlib.suppress(Exception):
+ router.propagation_transfer_state = router.PR_IDLE
+ ctx.config.lxmf_preferred_propagation_node_last_synced_at.set(
+ int(time.time())
+ )
+ await self.send_config_to_websocket_clients(context=ctx)
+ sync_start_time = None
+
# wait 1 second before next loop
await asyncio.sleep(1)
@@ -3290,6 +3343,10 @@ class ReticulumMeshChat:
with contextlib.suppress(Exception):
self._health_monitor.stop()
+ if self._mem_diag is not None:
+ with contextlib.suppress(Exception):
+ self._mem_diag.stop()
+
# force close websocket clients (copy: close() may touch the client list)
for websocket_client in list(self.websocket_clients):
try:
@@ -3539,6 +3596,120 @@ class ReticulumMeshChat:
},
)
+ # ── Memory diagnostics (only when --memory-diag is active) ──────────
+
+ @routes.get("/api/v1/diagnostics/memory")
+ async def get_memory_diagnostics(request):
+ if self._mem_diag is None:
+ return web.json_response(
+ {"enabled": False, "message": "Pass --memory-diag to enable"},
+ )
+ # tracemalloc.snapshot() + gc.get_objects() are CPU-bound and
+ # block the event loop for tens of seconds; run off-loop.
+ report = await asyncio.to_thread(self._mem_diag.report)
+ return web.json_response(report)
+
+ @routes.post("/api/v1/diagnostics/memory/snapshot")
+ async def take_memory_snapshot(request):
+ if self._mem_diag is None or not self._mem_diag.enabled:
+ return web.json_response(
+ {"error": "Memory diagnostics not enabled"},
+ status=400,
+ )
+ await asyncio.to_thread(self._mem_diag.snapshot)
+ gc_result = await asyncio.to_thread(self._mem_diag.find_cyclic_garbage)
+ stats = await asyncio.to_thread(self._mem_diag.gc_stats)
+ return web.json_response(
+ {
+ "status": "ok",
+ "snapshot_count": len(self._mem_diag._snapshots),
+ "gc_collected": gc_result,
+ "gc_stats": stats,
+ },
+ )
+
+ @routes.get("/api/v1/diagnostics/memory/heap")
+ async def get_heap_analysis(request):
+ if self._mem_diag is None or not self._mem_diag.enabled:
+ return web.json_response(
+ {"error": "Memory diagnostics not enabled"},
+ status=400,
+ )
+ top_n = int(request.query.get("top_n", 40))
+ by_type = await asyncio.to_thread(self._mem_diag.heap_by_type, top_n=top_n)
+ by_cat = await asyncio.to_thread(self._mem_diag.heap_by_category)
+ acc = await asyncio.to_thread(self._mem_diag.accumulating_types)
+ growth = await asyncio.to_thread(self._mem_diag.type_growth_since_start)
+ return web.json_response(
+ {
+ "by_type": by_type,
+ "by_category": by_cat,
+ "accumulating": acc,
+ "growth_since_start": growth,
+ },
+ )
+
+ @routes.get("/api/v1/diagnostics/memory/gc")
+ async def get_gc_stats(request):
+ if self._mem_diag is None or not self._mem_diag.enabled:
+ return web.json_response(
+ {"enabled": False, "message": "Pass --memory-diag to enable"},
+ )
+ stats = await asyncio.to_thread(self._mem_diag.gc_stats)
+ return web.json_response(stats)
+
+ @routes.post("/api/v1/diagnostics/memory/gc/collect")
+ async def force_gc_collect(request):
+ if self._mem_diag is None or not self._mem_diag.enabled:
+ return web.json_response(
+ {"error": "Memory diagnostics not enabled"},
+ status=400,
+ )
+ result = await asyncio.to_thread(self._mem_diag.find_cyclic_garbage)
+ if self._mem_diag.enabled:
+ await asyncio.to_thread(self._mem_diag.snapshot)
+ stats = await asyncio.to_thread(self._mem_diag.gc_stats)
+ return web.json_response(
+ {
+ "status": "ok",
+ "gc_collected": result,
+ "gc_stats": stats,
+ "snapshot_count": len(self._mem_diag._snapshots),
+ },
+ )
+
+ @routes.get("/api/v1/diagnostics/memory/referrers")
+ async def get_referrers(request):
+ if self._mem_diag is None or not self._mem_diag.enabled:
+ return web.json_response(
+ {"error": "Memory diagnostics not enabled"},
+ status=400,
+ )
+ type_name = request.query.get("type", "")
+ if not type_name:
+ return web.json_response(
+ {"error": "Specify ?type=<TypeName>"},
+ status=400,
+ )
+ result = await asyncio.to_thread(
+ self._mem_diag.find_referrers,
+ type_name,
+ )
+ return web.json_response(result)
+
+ @routes.post("/api/v1/diagnostics/memory/reset")
+ async def reset_memory_diagnostics(request):
+ if self._mem_diag is None:
+ return web.json_response(
+ {"error": "Memory diagnostics not enabled"},
+ status=400,
+ )
+ await asyncio.to_thread(self._mem_diag.reset)
+ await asyncio.to_thread(self._mem_diag.start)
+ return web.json_response({"status": "ok", "message": "Diagnostics reset"})
+
+ # ── Database ─────────────────────────────────────────────────────
+
@routes.post("/api/v1/database/snapshot")
async def create_db_snapshot(request):
try:
@@ -8068,6 +8239,9 @@ class ReticulumMeshChat:
if ident:
remote_identity_hash = ident.hash.hex()
+ if not remote_identity_hash:
+ # Fallback: use the provided lookup hash directly as identity hash
+ remote_identity_hash = lxmf_address or lxst_address
if not remote_identity_hash:
return web.json_response(
{"message": "Identity hash is required or could not be derived"},
@@ -12856,6 +13030,10 @@ class ReticulumMeshChat:
except Exception:
print("failed to launch web browser")
+ # start memory diagnostics periodic snapshot task
+ if self._mem_diag and self._mem_diag.enabled:
+ asyncio.create_task(self._memory_diag_snapshot_loop())
+
# create and run web app
app = web.Application(
client_max_size=1024 * 1024 * 50,
@@ -12935,6 +13113,17 @@ class ReticulumMeshChat:
protocol = "https" if use_https else "http"
print(f"Starting web server on {protocol}://{host}:{port}")
+ # Start memory diagnostics if enabled
+ if self._memory_diag_enabled:
+ from meshchatx.src.backend.diagnostics import MemoryDiagnostics
+
+ self._mem_diag = MemoryDiagnostics()
+ self._mem_diag.start()
+ print(
+ "[mem_diag] Memory diagnostics enabled — "
+ "see /api/v1/diagnostics/memory for reports",
+ )
+
if use_https and ssl_context:
web.run_app(app, host=host, port=port, ssl_context=ssl_context)
else:
@@ -13776,7 +13965,9 @@ class ReticulumMeshChat:
and self.telephone_manager
and self.telephone_manager.telephone
):
- self.telephone_manager.telephone.hangup()
+ self.telephone_manager.teardown()
+ elif value and self.telephone_manager:
+ self.telephone_manager.init_telephone()
if "telephone_allow_calls_from_contacts_only" in data:
self.config.telephone_allow_calls_from_contacts_only.set(
@@ -14873,6 +15064,10 @@ class ReticulumMeshChat:
self.websocket_clients.remove(client)
except ValueError:
pass
+ try:
+ await client.close(code=WSCloseCode.GOING_AWAY)
+ except Exception:
+ pass
# broadcasts config to all websocket clients
async def send_config_to_websocket_clients(self, context=None):
@@ -17781,6 +17976,13 @@ def main():
help="Clear the stored password hash on startup so a new password can be set via the web UI. Can also be set via MESHCHAT_RESET_PASSWORD environment variable.",
)
+ parser.add_argument(
+ "--memory-diag",
+ action="store_true",
+ default=env_bool("MESHCHAT_MEMORY_DIAG", False),
+ help="Enable tracemalloc-based memory diagnostics. Can also be set via MESHCHAT_MEMORY_DIAG environment variable.",
+ )
+
args = parser.parse_args()
ssl_cert = (args.ssl_cert or "").strip() or None
@@ -17914,6 +18116,7 @@ def main():
ssl_key_path=ssl_key,
rns_loglevel=rns_log_cli,
migration_context=migration_context,
+ memory_diag_enabled=args.memory_diag,
)
# store recovery on app for wiring with identity context
diff --git a/meshchatx/src/backend/database/provider.py b/meshchatx/src/backend/database/provider.py
index af40695a..124fccae 100644
--- a/meshchatx/src/backend/database/provider.py
+++ b/meshchatx/src/backend/database/provider.py
@@ -1,9 +1,14 @@
# SPDX-License-Identifier: 0BSD
import sqlite3
+import sys
import threading
import weakref
+_SQLITE_CONNECT_KW = {}
+if sys.version_info >= (3, 14):
+ _SQLITE_CONNECT_KW["cached_statements"] = 100
+
class DatabaseProvider:
_instance = None
@@ -43,6 +48,7 @@ class DatabaseProvider:
self.db_path,
check_same_thread=False,
isolation_level=None,
+ **_SQLITE_CONNECT_KW,
)
self._memory_connection.row_factory = sqlite3.Row
return self._memory_connection
@@ -54,6 +60,7 @@ class DatabaseProvider:
timeout=30.0,
check_same_thread=False,
isolation_level=None,
+ **_SQLITE_CONNECT_KW,
)
self._local.connection.row_factory = sqlite3.Row
# Enable WAL mode for better concurrency
diff --git a/meshchatx/src/backend/diagnostics/__init__.py b/meshchatx/src/backend/diagnostics/__init__.py
new file mode 100644
index 00000000..ff8b1dfd
--- /dev/null
+++ b/meshchatx/src/backend/diagnostics/__init__.py
@@ -0,0 +1,5 @@
+# SPDX-License-Identifier: 0BSD
+
+from .memory_diagnostics import MemoryDiagnostics, get_diagnostics, take_heap_snapshot
+
+__all__ = ["MemoryDiagnostics", "get_diagnostics", "take_heap_snapshot"]
diff --git a/meshchatx/src/backend/diagnostics/memory_diagnostics.py b/meshchatx/src/backend/diagnostics/memory_diagnostics.py
new file mode 100644
index 00000000..af9ebc43
--- /dev/null
+++ b/meshchatx/src/backend/diagnostics/memory_diagnostics.py
@@ -0,0 +1,529 @@
+import gc
+import sys
+import tracemalloc
+from typing import Any, Optional
+
+
+def _obj_size(obj: Any) -> int:
+ try:
+ return sys.getsizeof(obj)
+ except Exception:
+ return 0
+
+
+def _safe_type_name(obj: Any) -> str:
+ try:
+ return type(obj).__qualname__
+ except Exception:
+ return "<unknown>"
+
+
+def _safe_module(obj: Any) -> str:
+ try:
+ return type(obj).__module__
+ except Exception:
+ return ""
+
+
+def _is_reticulum_obj(obj: Any) -> bool:
+ mod = _safe_module(obj)
+ return "RNS" in mod or "LXMF" in mod or "LXST" in mod or "meshchatx" in mod
+
+
+def _is_closure(obj: Any) -> bool:
+ return type(obj).__name__ in ("function", "cell", "code", "FrameType", "MethodType")
+
+
+def _classify(obj: Any) -> str:
+ t = type(obj)
+ if t in (str, bytes, bytearray, int, float, bool, type(None)):
+ return "builtin"
+ if isinstance(obj, (list, tuple, set, frozenset)):
+ return "container"
+ if isinstance(obj, dict):
+ return "dict"
+ mod = _safe_module(obj)
+ if "RNS" in mod:
+ return "RNS"
+ if "LXMF" in mod:
+ return "LXMF"
+ if "LXST" in mod:
+ return "LXST"
+ if "meshchatx" in mod:
+ return "meshchatx"
+ if mod.startswith(("asyncio", "concurrent", "threading")):
+ return "async"
+ return "other"
+
+
+class _ObjectTypeTracker:
+ """Tracks per-type object counts across snapshots to detect accumulation."""
+
+ def __init__(self) -> None:
+ self._history: list[dict[str, int]] = []
+ self._last_full: dict[str, int] = {}
+
+ def record(self) -> None:
+ counts: dict[str, int] = {}
+ try:
+ for obj in gc.get_objects():
+ try:
+ name = _safe_type_name(obj)
+ counts[name] = counts.get(name, 0) + 1
+ except Exception:
+ continue
+ except Exception:
+ pass
+ self._history.append(counts)
+ self._last_full = counts
+
+ @property
+ def growth_since_first(self) -> list[tuple[str, int]]:
+ if len(self._history) < 2:
+ return []
+ first = self._history[0]
+ last = self._history[-1]
+ result: list[tuple[str, int]] = []
+ for tname, count in last.items():
+ diff = count - first.get(tname, 0)
+ if diff > 0:
+ result.append((tname, diff))
+ result.sort(key=lambda x: -x[1])
+ return result
+
+ @property
+ def accumulating_types(self) -> list[tuple[str, int]]:
+ if len(self._history) < 3:
+ return []
+ result: list[tuple[str, int]] = []
+ for tname in self._history[-1]:
+ counts = [h.get(tname, 0) for h in self._history]
+ if (
+ all(counts[i] <= counts[i + 1] for i in range(len(counts) - 1))
+ and counts[-1] - counts[0] > 0
+ ):
+ result.append((tname, counts[-1] - counts[0]))
+ result.sort(key=lambda x: -x[1])
+ return result
+
+
+class MemoryDiagnostics:
+ """tracemalloc-based memory diagnostics for identifying leaks.
+
+ Python 3.14+ uses an incremental GC that spreads collection work across
+ allocations instead of doing one big sweep. Cyclic garbage can therefore
+ linger longer between full collections. This class helps detect
+ accumulating objects by:
+
+ * Taking periodic ``tracemalloc`` snapshots and diffing against a baseline.
+ * Tracking GC generation sizes (gen0/gen1/gen2 object counts).
+ * Profiling ``gc.get_objects()`` by type to spot monotonic growth.
+ * Finding the top-N allocation sites (filename + line number) that
+ contribute the most to memory growth.
+
+ Usage::
+
+ from meshchatx.src.backend.diagnostics import MemoryDiagnostics
+ diag = MemoryDiagnostics()
+ diag.start()
+ # ... let app run for a while ...
+ report = diag.report()
+ diag.stop()
+
+ All methods are safe to call when tracing is inactive (they return
+ empty/default values).
+ """
+
+ def __init__(self, nframes: int = 25) -> None:
+ self._nframes = nframes
+ self._enabled = False
+ self._baseline: Optional[tracemalloc.Snapshot] = None
+ self._snapshots: list[tracemalloc.Snapshot] = []
+ self._gc_stats: list[dict[str, Any]] = []
+ self._type_tracker = _ObjectTypeTracker()
+
+ # ------------------------------------------------------------------
+ # Lifecycle
+ # ------------------------------------------------------------------
+
+ def start(self) -> None:
+ if self._enabled:
+ return
+ if not tracemalloc.is_tracing():
+ tracemalloc.start(self._nframes)
+ self._enabled = True
+ self._baseline = tracemalloc.take_snapshot()
+ self._snapshots = [self._baseline]
+ self._record_gc_stats()
+ self._type_tracker.record()
+ print(
+ f"[mem_diag] Started tracing (nframes={self._nframes}, "
+ f"gc.freeze={getattr(gc, 'freeze_count', 0)})",
+ )
+
+ def stop(self) -> None:
+ if not self._enabled:
+ return
+ if tracemalloc.is_tracing():
+ tracemalloc.stop()
+ self._enabled = False
+ print("[mem_diag] Stopped tracing")
+
+ def reset(self) -> None:
+ self.stop()
+ self._baseline = None
+ self._snapshots.clear()
+ self._gc_stats.clear()
+
+ @property
+ def enabled(self) -> bool:
+ return self._enabled and tracemalloc.is_tracing()
+
+ # ------------------------------------------------------------------
+ # Snapshots
+ # ------------------------------------------------------------------
+
+ def snapshot(self) -> Optional[tracemalloc.Snapshot]:
+ if not self.enabled:
+ return None
+ snap = tracemalloc.take_snapshot()
+ self._snapshots.append(snap)
+ self._record_gc_stats()
+ self._type_tracker.record()
+ return snap
+
+ def diff_snapshots(
+ self,
+ new_idx: int = -1,
+ old_idx: int = 0,
+ key_type: str = "lineno",
+ top_n: int = 40,
+ ) -> list[dict[str, Any]]:
+ if not self._snapshots or len(self._snapshots) < 2:
+ return []
+ if old_idx < 0:
+ old_idx = len(self._snapshots) + old_idx
+ if new_idx < 0:
+ new_idx = len(self._snapshots) + new_idx
+ if old_idx < 0 or old_idx >= len(self._snapshots):
+ return []
+ if new_idx < 0 or new_idx >= len(self._snapshots):
+ return []
+ diff = self._snapshots[new_idx].compare_to(self._snapshots[old_idx], key_type)
+ result: list[dict[str, Any]] = []
+ for stat in diff[:top_n]:
+ frame = stat.traceback[0]
+ result.append(
+ {
+ "size": stat.size,
+ "size_mib": round(stat.size / (1024 * 1024), 3),
+ "count": stat.count,
+ "file": frame.filename,
+ "line": frame.lineno,
+ },
+ )
+ return result
+
+ def top_allocation_sites(self, top_n: int = 40) -> list[dict[str, Any]]:
+ if not self.enabled:
+ return []
+ snap = tracemalloc.take_snapshot()
+ stats = snap.statistics("lineno", True)
+ result: list[dict[str, Any]] = []
+ for stat in stats[:top_n]:
+ frame = stat.traceback[0]
+ result.append(
+ {
+ "size": stat.size,
+ "size_mib": round(stat.size / (1024 * 1024), 3),
+ "count": stat.count,
+ "file": frame.filename,
+ "line": frame.lineno,
+ },
+ )
+ return result
+
+ # ------------------------------------------------------------------
+ # GC generation tracking (critical for incremental GC diagnostics)
+ # ------------------------------------------------------------------
+
+ def _record_gc_stats(self) -> None:
+ g0 = g1 = g2 = -1
+ try:
+ g0 = len(gc.get_objects(0))
+ g1 = len(gc.get_objects(1))
+ g2 = len(gc.get_objects(2))
+ except TypeError:
+ g0 = g1 = g2 = -1
+ except Exception:
+ pass
+ self._gc_stats.append(
+ {
+ "gen0": g0,
+ "gen1": g1,
+ "gen2": g2,
+ "total": sum(x for x in (g0, g1, g2) if x >= 0),
+ "thresholds": list(gc.get_threshold()),
+ "count": gc.get_count(),
+ "frozen": getattr(gc, "freeze_count", 0),
+ },
+ )
+
+ def gc_stats(self) -> dict[str, Any]:
+ if not self._gc_stats:
+ return {"available": False}
+ first = self._gc_stats[0]
+ last = self._gc_stats[-1]
+ return {
+ "available": True,
+ "snapshots_taken": len(self._snapshots),
+ "records": len(self._gc_stats),
+ "first": first,
+ "last": last,
+ "deltas": {
+ k: last.get(k, 0) - first.get(k, 0)
+ for k in ("gen0", "gen1", "gen2", "total")
+ if k in first and k in last
+ },
+ "history": self._gc_stats,
+ }
+
+ # ------------------------------------------------------------------
+ # Heap analysis (gc.get_objects based)
+ # ------------------------------------------------------------------
+
+ def heap_by_type(self, top_n: int = 40) -> list[dict[str, Any]]:
+ """Count live objects grouped by their type name."""
+ counts: dict[str, int] = {}
+ sizes: dict[str, int] = {}
+ try:
+ for obj in gc.get_objects():
+ try:
+ tname = _safe_type_name(obj)
+ counts[tname] = counts.get(tname, 0) + 1
+ sizes[tname] = sizes.get(tname, 0) + _obj_size(obj)
+ except Exception:
+ continue
+ except Exception:
+ return []
+ sorted_types = sorted(counts.items(), key=lambda x: -x[1])
+ return [
+ {
+ "type": tname,
+ "count": cnt,
+ "size_bytes": sizes.get(tname, 0),
+ "size_mib": round(sizes.get(tname, 0) / (1024 * 1024), 3),
+ }
+ for tname, cnt in sorted_types[:top_n]
+ ]
+
+ def heap_by_category(self, top_n: int = 40) -> list[dict[str, Any]]:
+ """Count live objects grouped by category (RNS, LXMF, meshchatx, etc)."""
+ counts: dict[str, int] = {}
+ sizes: dict[str, int] = {}
+ try:
+ for obj in gc.get_objects():
+ try:
+ cat = _classify(obj)
+ counts[cat] = counts.get(cat, 0) + 1
+ sizes[cat] = sizes.get(cat, 0) + _obj_size(obj)
+ except Exception:
+ continue
+ except Exception:
+ return []
+ sorted_cats = sorted(counts.items(), key=lambda x: -x[1])
+ return [
+ {
+ "category": cat,
+ "count": cnt,
+ "size_bytes": sizes.get(cat, 0),
+ "size_mib": round(sizes.get(cat, 0) / (1024 * 1024), 3),
+ }
+ for cat, cnt in sorted_cats[:top_n]
+ ]
+
+ def accumulating_types(self, top_n: int = 20) -> list[tuple[str, int]]:
+ return self._type_tracker.accumulating_types[:top_n]
+
+ def type_growth_since_start(self, top_n: int = 20) -> list[tuple[str, int]]:
+ return self._type_tracker.growth_since_first[:top_n]
+
+ # ------------------------------------------------------------------
+ # Reference cycle detection helpers
+ # ------------------------------------------------------------------
+
+ def find_referrers(
+ self, type_name: str, max_results: int = 20
+ ) -> list[dict[str, Any]]:
+ """Find referrers of objects of a given type — useful to trace who holds references."""
+ matches: list[Any] = []
+ try:
+ for obj in gc.get_objects():
+ try:
+ if _safe_type_name(obj) == type_name:
+ matches.append(obj)
+ if len(matches) >= max_results:
+ break
+ except Exception:
+ continue
+ except Exception:
+ return []
+ result: list[dict[str, Any]] = []
+ for obj in matches:
+ refs: list[str] = []
+ try:
+ for ref in gc.get_referrers(obj):
+ try:
+ refs.append(f"{_safe_type_name(ref)} at {id(ref):#x}")
+ except Exception:
+ refs.append("<error>")
+ if len(refs) >= 10:
+ break
+ except Exception:
+ refs = ["<error>"]
+ result.append(
+ {
+ "type": _safe_type_name(obj),
+ "id": id(obj),
+ "size": _obj_size(obj),
+ "referrers": refs,
+ },
+ )
+ return result
+
+ def find_cyclic_garbage(self) -> list[dict[str, Any]]:
+ """Run gc.collect() and return info about what was collected."""
+ unreachable_before = gc.get_count()
+ cols: list[int] = []
+ for gen in range(3):
+ try:
+ n = gc.collect(gen)
+ cols.append(n)
+ except Exception:
+ cols.append(-1)
+ unreachable_after = gc.get_count()
+ return [
+ {
+ "generation": i,
+ "collected": n,
+ }
+ for i, n in enumerate(cols)
+ ] + [
+ {
+ "unreachable_before": list(unreachable_before),
+ "unreachable_after": list(unreachable_after),
+ },
+ ]
+
+ def gc_garbage_types(self) -> list[dict[str, int]]:
+ """Show types of objects in gc.garbage (uncollectable objects)."""
+ counts: dict[str, int] = {}
+ try:
+ for obj in gc.garbage:
+ try:
+ tname = _safe_type_name(obj)
+ counts[tname] = counts.get(tname, 0) + 1
+ except Exception:
+ continue
+ except Exception:
+ pass
+ sorted_counts = sorted(counts.items(), key=lambda x: -x[1])
+ return [{"type": t, "count": c} for t, c in sorted_counts]
+
+ # ------------------------------------------------------------------
+ # Consolidated report
+ # ------------------------------------------------------------------
+
+ def report(self) -> dict[str, Any]:
+ """Generate a comprehensive memory diagnostics report."""
+ if not self._enabled:
+ return {
+ "enabled": False,
+ "message": "Memory diagnostics not enabled (pass --memory-diag)",
+ }
+
+ current_traced, peak_traced = (0, 0)
+ if tracemalloc.is_tracing():
+ current_traced, peak_traced = tracemalloc.get_traced_memory()
+
+ growth = self.diff_snapshots(top_n=30)
+ top_sites = self.top_allocation_sites(top_n=15)
+ gc_info = self.gc_stats()
+ top_types = self.heap_by_type(top_n=30)
+ cat_types = self.heap_by_category()
+ accumulating = self.accumulating_types()
+ growth_types = self.type_growth_since_start()
+
+ return {
+ "enabled": True,
+ "python_version": sys.version,
+ "incremental_gc": self._detect_incremental_gc(),
+ "tracemalloc": {
+ "current_bytes": current_traced,
+ "current_mib": round(current_traced / (1024 * 1024), 3),
+ "peak_bytes": peak_traced,
+ "peak_mib": round(peak_traced / (1024 * 1024), 3),
+ },
+ "gc": gc_info,
+ "growth_vs_baseline": growth,
+ "top_allocation_sites": top_sites,
+ "heap_by_type": top_types,
+ "heap_by_category": cat_types,
+ "accumulating_types": [{"type": t, "growth": c} for t, c in accumulating],
+ "type_growth_since_start": [
+ {"type": t, "growth": c} for t, c in growth_types
+ ],
+ "gc_freeze_count": getattr(gc, "freeze_count", 0),
+ "snapshot_count": len(self._snapshots),
+ }
+
+ @staticmethod
+ def _detect_incremental_gc() -> dict[str, Any]:
+ """Detect if we're running on Python 3.14+ with incremental GC."""
+ version_info = sys.version_info
+ result: dict[str, Any] = {
+ "is_incremental_gc": version_info >= (3, 14),
+ "python_version": f"{version_info.major}.{version_info.minor}.{version_info.micro}",
+ }
+ if version_info >= (3, 14):
+ thresh = gc.get_threshold()
+ result["thresholds"] = list(thresh)
+ result["note"] = (
+ "Python 3.14+ uses incremental GC. "
+ "Full collections may be deferred, causing cyclic garbage "
+ "to linger. Check 'gc.deltas.gen2' growth."
+ )
+ return result
+
+
+# ------------------------------------------------------------------
+# Module-level convenience
+# ------------------------------------------------------------------
+
+_diag: Optional[MemoryDiagnostics] = None
+
+
+def get_diagnostics() -> MemoryDiagnostics:
+ global _diag
+ if _diag is None:
+ _diag = MemoryDiagnostics()
+ return _diag
+
+
+def take_heap_snapshot(include_tracemalloc: bool = True) -> dict[str, Any]:
+ """Convenience function: take a one-shot heap snapshot.
+
+ Useful for ``pdb`` / ``breakpoint()`` sessions::
+
+ from meshchatx.src.backend.diagnostics import take_heap_snapshot
+ report = take_heap_snapshot()
+ """
+ diag = get_diagnostics()
+ was_enabled = diag.enabled
+ if not was_enabled and include_tracemalloc:
+ diag.start()
+ diag.snapshot()
+ report = diag.report()
+ if not was_enabled and include_tracemalloc:
+ diag.stop()
+ return report
diff --git a/meshchatx/src/backend/recovery/health_monitor.py b/meshchatx/src/backend/recovery/health_monitor.py
index 93d2ca14..6b14ebea 100644
--- a/meshchatx/src/backend/recovery/health_monitor.py
+++ b/meshchatx/src/backend/recovery/health_monitor.py
@@ -15,6 +15,7 @@ import collections
import gc
import json
import logging
+import sys
import threading
import psutil
@@ -60,7 +61,10 @@ class HealthMonitor:
self._check()
except Exception as exc:
_log.debug("HealthMonitor check error: %s", exc)
- gc.collect()
+ if sys.version_info >= (3, 14) and gc.get_threshold()[2] == 0:
+ gc.collect(2)
+ else:
+ gc.collect()
await asyncio.sleep(self.CHECK_INTERVAL)
def _check(self):
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────